package org.joy.hpc.analyzer;

import java.io.File;
import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.joy.analyzer.Hit;
import org.joy.db.Env;
import org.joy.db.IndexDB;
import org.joy.db.IndexEntry;

public class HitReducer extends MapReduceBase implements
		Reducer<Text, HitWritable, Text, HitWritable> {

	@Override
	public void reduce(Text term, Iterator<HitWritable> vals,
			OutputCollector<Text, HitWritable> out, Reporter reporter)
			throws IOException {
		// 把每一个Reduce Group所有的hits从中间结果取出
		ArrayList<HitWritable> hitsLst = new ArrayList<HitWritable>();
		while (vals.hasNext()) {
			HitWritable h = vals.next();
			hitsLst.add(new HitWritable(h));
		}
		// 按照关键字权重去排序
		Collections.sort(hitsLst);
		// 写入输出格式，在这里只要更改少部分代码就可以实现真正的倒排索引
//		DecimalFormat format = new DecimalFormat(".00");
//		for (HitWritable h : hitsLst) {
//			String pos = "";
//			for (Writable w : h.getPos()) {
//				pos += ((IntWritable) w).get() + "		";
//			}
//			out.collect(new Text(term), new Text(h.getUrl() + "\t"
//					+ format.format(h.getScore()) + "\t " + pos));
//		}
		// 输出到真正的倒排索引中去
		for (HitWritable h : hitsLst) {
			out.collect(term, h);
		}
	}

	@Override
	public void configure(JobConf job) {
		try {
			File folder = new File(Env.DB_FOLDER);
			if (!folder.exists())
				folder.mkdir();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		// TODO Auto-generated method stub
		super.configure(job);
	}
}
